package org.cloud.automatic.queue.job;

import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import lombok.extern.slf4j.Slf4j;
import org.cloud.automatic.queue.config.SystemSendQueueConfig;
import org.cloud.automatic.queue.entity.SendQueueWorkerBO;
import org.cloud.automatic.queue.entity.SystemSendQueueBO;
import org.cloud.automatic.queue.service.SystemSendQueueBOService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.List;

/**
 * @Description: 补偿自动实现
 * @Author: gangzhenga
 * @Date: 2021/7/28 11:30
 */
@Slf4j
@Service("transactionRetryService")
public class TransactionRetryServiceImpl {

	@Autowired
	private SystemSendQueueBOService systemSendQueueBOService;

	//@Autowired
	//private RedisUtil redisUtil;

	@Autowired
	private SystemSendQueueConfig systemSendQueueConfig;

	/**
	 * 补偿时间间隔策略定义
	 * 如果超出最长时间间隔，则不再补偿，只能通过手动干预的方式重新执行
	 */
	private static final int[] RESEN_TIMES = new int[]{0,1,5,15,30,60,120,360,720,1440,2160,2880};


	/**
	 * 项目启动1分钟后开始执行，每分钟执行一次
	 */
	@Scheduled(initialDelay=60000, fixedRate=60000)
	public void startTask(){

		if(!systemSendQueueConfig.isEnable()){
			//log.info("---------事务补偿，开关未开启，停止补偿-----------");
			return;
		}

		log.info("---------事务补偿，TransactionRetryServiceImpl.startTask--------begin----");

		SendQueueMaster master = null;

		/**
		 * 1、启动补偿任务处理器，根据当前CPU核心数，开启任务执行线程，并等待接收任务
		 * 2、根据自动任务配置的分片执行标识，读取当前服务节点产生的事务补偿点，每次读取1000条；
		 *        循环处理：
		 *             如果 查询列表大小为0，则通知补偿任务执行器，在队列中的任务全部执行完成后可以销毁执行器资源
		 *             任务校验，未超出最大补偿次数的和已达到本次补偿时间的任务才能进入补偿队列
		 *             如果符合补偿条件，则对任务加锁，并提交至任务队列
		 *
		 * 3、开启while循环等待队列中任务全部执行完成，通知任务处理器终止所有线程，补偿结束
		 */

		try{

			// 标记本次未完成
			boolean sendflag = true;

			// 根据cpu核心启动任务补偿器
			int availableProcessors = Runtime.getRuntime().availableProcessors() - 1 ;
			master = new SendQueueMaster(new SendQueueWorker(), availableProcessors);
			master.executes();

			// 查询发送和发送失败的队列信息（每次查1000条）
			Page<SystemSendQueueBO> page = new Page<>(1, 1000);
			SystemSendQueueBO systemSendQueueBO = SystemSendQueueBO.builder().moduleFlag(systemSendQueueConfig.getModuleFlag()).build();

			while(sendflag){

				IPage<SystemSendQueueBO> systemSendQueueBOIPage = systemSendQueueBOService.queryPage(page, systemSendQueueBO);

				/**
				 * 没有结果则终止本次调度
				 * master通知各线程，如果队列内的任务执行完成，可以终止线程
				 */
				if (systemSendQueueBOIPage == null || systemSendQueueBOIPage.getRecords().size() == 0){
					sendflag = false;
					master.noticeCanStopWorker();
					break;
				}

				List<SystemSendQueueBO> systemSendQueueBOList = systemSendQueueBOIPage.getRecords();

				for(SystemSendQueueBO newSystemSendQueueBO : systemSendQueueBOList){

					if(newSystemSendQueueBO == null || newSystemSendQueueBO.getSendNum() == null){
						continue ;
					}

					if(newSystemSendQueueBO.getSendNum().intValue() >= RESEN_TIMES.length){
						continue ;
					}

					/**
					 * 进入补偿队列之前，对数据加队列并发锁，避免并发补偿；
					 * 此处只负责加锁，依赖于redis超时时间自动解锁
					 */
					//boolean lock = redisUtil.tryLock("queue-lock-" + newSystemSendQueueBO.getBizId());
					boolean lock = false;
					if(lock){
						SendQueueWorkerBO workBO = new SendQueueWorkerBO();
						workBO.setData(newSystemSendQueueBO);
						master.submitTask(workBO);
					}
				}

				// 设置下一页
				page.setCurrent(page.getCurrent() + 1);
			}

			// 等待队列内的任务执行完成
			while(true){
				if(master.getWorkQueue().size() == 0){
					break;
				}
			}


		}catch (Exception e){
			log.error("===事务补偿出现异常===", e);
		}finally {
			log.info("---------事务补偿，TransactionRetryServiceImpl.startTask--------end----");
		}

	}
}
